package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.EnergySocBean;
import com.zhike.function.BeanToJsonFunction;
import com.zhike.function.DimAsyncFunction;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhike
 * @CreateTime: 2024/3/5 14:20
 * @Description: 计算各车每次行程的soc消耗，行驶里程
 */
public class DwsEnergySocWindow {
    public static void main(String[] args) {


        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(1);

        //2.读取car_electricity_log日志创建流并转换为EnergySocBean对象
        SingleOutputStreamOperator<EnergySocBean> eleDS = env.fromSource(KafkaUtils.getKafkaSource("car_electricity_log", "dws_energy_soc_window_1")
                        , WatermarkStrategy.noWatermarks(), "car_electricity_log")
                .map((MapFunction<String, EnergySocBean>) DwsEnergySocWindow::getEnergySocBean);

        //3.读取car_hybrid_log日志创建流并转换为EnergySocBean对象
        SingleOutputStreamOperator<EnergySocBean> hybDS = env.fromSource(KafkaUtils.getKafkaSource("car_hybrid_log", "dws_energy_soc_window_2")
                        , WatermarkStrategy.noWatermarks(), "car_hybrid_log")
                .map((MapFunction<String, EnergySocBean>) DwsEnergySocWindow::getEnergySocBean);

        //4.读取car_fuel_log日志创建流并转换为EnergySocBean对象
        SingleOutputStreamOperator<EnergySocBean> fuelDS = env.fromSource(KafkaUtils.getKafkaSource("car_fuel_log", "dws_energy_soc_window_3")
                        , WatermarkStrategy.noWatermarks(), "car_fuel_log")
                .map((MapFunction<String, EnergySocBean>) DwsEnergySocWindow::getEnergySocBean);

        //5.合并流创建
        DataStream<EnergySocBean> unionDS = eleDS.union(hybDS, fuelDS);

        //6.根据vin分组计算行程id
        SingleOutputStreamOperator<EnergySocBean> tripDS = unionDS.keyBy(new KeySelector<EnergySocBean, String>() {
            @Override
            public String getKey(EnergySocBean value) throws Exception {
                return value.getVin();
            }
        }).map(new RichMapFunction<EnergySocBean, EnergySocBean>() {
            private ValueState<Integer> tripIdState;
            private ValueState<Long> tsState;

            @Override
            public void open(Configuration parameters) throws Exception {
                tsState = getRuntimeContext().getState(new ValueStateDescriptor<>("ts_state", Long.class));
                tripIdState = getRuntimeContext().getState(new ValueStateDescriptor<>("trip_id_state", Integer.class));
            }

            @Override
            public EnergySocBean map(EnergySocBean value) throws Exception {
                //获取上一次行程
                Integer tripID = tripIdState.value();
                if (tripID == null) {
                    tripID = 1;
                }
                //获取上条日志时间戳
                Long tsPre = tsState.value();
                //获取当前日志时间戳
                Long ts = value.getTs();
                if (tsPre == null) {
                    tsState.update(ts);
                    value.setTripId(tripID);
                } else if (ts - tsPre > 60000) {
                    tripIdState.update(tripID + 1);
                    tsState.update(ts);
                    value.setTripId(tripID + 1);
                } else {
                    tsState.update(ts);
                    value.setTripId(tripID);
                }
                return value;
            }
        });

        //7.设置水位线
        SingleOutputStreamOperator<EnergySocBean> wMDS = tripDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<EnergySocBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                .withTimestampAssigner(
                        (SerializableTimestampAssigner<EnergySocBean>) (element, recordTimestamp) -> element.getTs()));

        //8.根据vin,行程id分组
        KeyedStream<EnergySocBean, Tuple2<String, Integer>> keyByDS = wMDS.keyBy(new KeySelector<EnergySocBean, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> getKey(EnergySocBean value) throws Exception {
                return new Tuple2<>(value.getVin(), value.getTripId());
            }
        });

        //9.开窗，聚合
        SingleOutputStreamOperator<EnergySocBean> reduceDS = keyByDS.window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<EnergySocBean>() {
                    @Override
                    public EnergySocBean reduce(EnergySocBean value1, EnergySocBean value2) throws Exception {
                        value1.setMaxMileage(Math.max(value1.getMaxMileage(), value2.getMaxMileage()));
                        value1.setMinMileage(Math.min(value1.getMinMileage(), value2.getMinMileage()));
                        value1.setMinSoc(Math.min(value1.getMaxSoc(), value2.getMaxSoc()));
                        value1.setMaxSoc(Math.max(value1.getMinSoc(), value2.getMinSoc()));
                        value1.setMaxTs(Math.max(value1.getMaxTs(), value2.getMaxTs()));
                        value1.setMinTs(Math.min(value1.getMinTs(), value2.getMinTs()));
                        return value1;
                    }
                }, new WindowFunction<EnergySocBean, EnergySocBean, Tuple2<String, Integer>, TimeWindow>() {
                    @Override
                    public void apply(Tuple2<String, Integer> stringIntegerTuple2, TimeWindow window, Iterable<EnergySocBean> input, Collector<EnergySocBean> out) throws Exception {
                        EnergySocBean bean = input.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });


        //10.关联维度表补充车型，品牌
        SingleOutputStreamOperator<EnergySocBean> rishDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<EnergySocBean>("DIM_CAR_INFO") {
                    @Override
                    public String getKey(EnergySocBean input) {
                        return input.getVin();
                    }

                    @Override
                    public void join(EnergySocBean input, JSONObject dimInfo) {
                        input.setCarType(dimInfo.getString("TYPE"));
                        input.setBrand(dimInfo.getString("TRADEMARK"));
                    }
                }
                , 100L, TimeUnit.SECONDS);

        rishDS.print();
        //11.将结果写入doris

        SingleOutputStreamOperator<String> resultDS = rishDS.map(BeanToJsonFunction::getString);
//        resultDS.print();
        resultDS.sinkTo(DorisUtils.getDorisSink("dws_energy_soc_window"));

        //12.启动任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static EnergySocBean getEnergySocBean(String value) {
        JSONObject json = JSON.parseObject(value);
        //获取vin
        String vin = json.getString("vin");
        //获取soc
        Integer soc = json.getInteger("soc");
        //获取mileage
        Double mileage = json.getDouble("mileage");
        //获取时间戳
        Long ts = json.getLong("timestamp");
        return EnergySocBean.builder()
                .vin(vin)
                .maxMileage(mileage)
                .minMileage(mileage)
                .maxSoc(soc)
                .minSoc(soc)
                .maxTs(ts)
                .minTs(ts)
                .ts(ts)
                .build();
    }
}
